package demo.base;

import org.apache.flink.api.java.utils.ParameterTool;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import demo.base.connect.conf.PropertyReader;

import java.io.IOException;
import java.util.Set;

import static com.alibaba.fastjson.serializer.SerializerFeature.WriteMapNullValue;

public class FlinkParamterToolDemo {
    public static void main(String[] args) throws IOException {
        //        InputStream resourceAsStream =
        // ParameterTool.class.getClassLoader().getResourceAsStream("userlabel_prod.properties");
        //        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(resourceAsStream);
        //
        ////        System.out.println(parameterTool.get("KFKSERVER"));
        //        System.out.println(parameterTool.getNumberOfParameters());
        //        Set<String> keys = parameterTool.getConfiguration().keySet();
        //
        //
        //        System.out.println(keys);

        JSONObject jsonObject = initJsonObjState();
        System.out.println(JSON.toJSONString(jsonObject, WriteMapNullValue));
    }

    private static JSONObject initJsonObjState() {
        JSONObject res = new JSONObject();
        try {
            ParameterTool reader = PropertyReader.getReader("order.properties");
            Set<Object> propSets = reader.getProperties().keySet();

            for (Object key : propSets) {
                String typeValue = reader.get((String) key);
                String[] splits = typeValue.split(",");
                String type = splits[0];
                String value = splits[1];
                JsonObjectSetValue(res, type, (String) key, value);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return res;
    }

    private static void JsonObjectSetValue(
            JSONObject jsonObj, String type, String key, String value) {
        if (!value.equals("null")) {
            switch (type) {
                case "Long":
                    jsonObj.put(key, Long.parseLong(value));
                    break;
                case "String":
                    jsonObj.put(key, value);
                    break;
                case "Double":
                    jsonObj.put(key, Double.parseDouble(value));
                    break;
                case "Integer":
                    jsonObj.put(key, Integer.parseInt(value));
                    break;
                default:
                    System.out.println("Not support this type now!");
                    break;
            }
        } else {
            jsonObj.put(key, null);
        }
    }
}
